package com.factors.DaiGuiCheng.Sleep_Hours;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class ExamScoreSleepMapper extends Mapper<LongWritable, Text, Text, DoubleWritable> {

    private static final String RANGE_LOW = "0-59";
    private static final String RANGE_MEDIUM_LOW = "60-79";
    private static final String RANGE_MEDIUM_HIGH = "80-89";
    private static final String RANGE_HIGH = "90-100";

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        if (key.get() == 0) return; // Skip header line

        String[] columns = value.toString().split(",");
        if (columns.length >= 20) {
            try {
                float examScore = Float.parseFloat(columns[19]); // Exam_Score is the 20th column
                float sleepHours = Float.parseFloat(columns[5]); // Sleep_Hours is the 6th column

                // Determine the score range
                String scoreRange;
                if (examScore >= 0 && examScore <= 59) {
                    scoreRange = RANGE_LOW;
                } else if (examScore >= 60 && examScore <= 79) {
                    scoreRange = RANGE_MEDIUM_LOW;
                } else if (examScore >= 80 && examScore <= 89) {
                    scoreRange = RANGE_MEDIUM_HIGH;
                } else if (examScore >= 90 && examScore <= 100) {
                    scoreRange = RANGE_HIGH;
                } else {
                    // If exam score is outside of expected ranges, skip this record
                    return;
                }

                // Write the score range and sleep hours to the context
                context.write(new Text(scoreRange), new DoubleWritable(sleepHours));
            } catch (NumberFormatException e) {
                // Handle non-numeric values or other parsing errors
            }
        }
    }
}